package com.afdor.rws.event.driven.support;

import com.afdor.rws.annotation.extension.Extension;
import com.afdor.rws.core.Callback;
import com.afdor.rws.core.exception.BusinessException;
import com.afdor.rws.core.extension.ExtensionLoader;
import com.afdor.rws.core.strategy.IdGeneratorStrategy;
import com.afdor.rws.event.driven.api.EventBus;
import com.afdor.rws.event.driven.api.EventChangeManager;
import com.afdor.rws.event.driven.api.EventPersistenceManager;
import com.afdor.rws.event.driven.api.channel.consumer.EventConsumer;
import com.afdor.rws.event.driven.api.channel.consumer.handler.EventHandler;
import com.afdor.rws.event.driven.api.channel.consumer.handler.EventRetryStrategy;
import com.afdor.rws.event.driven.api.channel.producer.EventProducer;
import com.afdor.rws.event.driven.api.channel.producer.transaction.EventProducerTransactionItem;
import com.afdor.rws.event.driven.api.transaction.EventBusTransactionItem;
import com.afdor.rws.event.driven.config.EventBusConfig;
import com.afdor.rws.event.driven.config.EventConsumerConfig;
import com.afdor.rws.event.driven.config.EventProducerConfig;
import com.afdor.rws.event.driven.entities.AggregationEvent;
import com.afdor.rws.event.driven.entities.Event;
import com.afdor.rws.event.driven.exception.EventSendException;
import com.afdor.rws.event.driven.support.holder.CallbackEvent;
import com.afdor.rws.event.driven.support.holder.EventResult;
import lombok.Getter;
import lombok.NonNull;
import lombok.extern.slf4j.Slf4j;

import java.util.HashMap;
import java.util.List;
import java.util.Objects;
import java.util.Optional;
import java.util.Queue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicMarkableReference;
import java.util.stream.Collectors;

/**
 * 事件总线默认实现
 *
 * @author 悭梵
 * @date Created in 2018-12-26 14:18
 */
@Slf4j
@Extension(value = "default", order = Extension.NORMAL_ORDER - 100, interfaceClass = EventBus.class)
public class DefaultEventBus implements EventBus {
    /**
     * 初始化状态
     */
    private final AtomicBoolean init = new AtomicBoolean();

    /**
     * 配置信息
     */
    @Getter
    private EventBusConfig config;
    /**
     * 持久化管理器
     */
    private EventPersistenceManager eventPersistenceManager;
    /**
     * 事件窗台变更管理器
     */
    private EventChangeManager eventStatusChangeManager;
    /**
     * 执行通道配置信息
     */
    @Getter
    private final ConcurrentMap<String, ChannelProvider> channelProviderMap = new ConcurrentHashMap<>(32);
    /**
     * 事物提交队列管理器
     */
    @Getter
    private ConcurrentMap<String, EventBusTransactionItem> eventTransactionMap = new ConcurrentHashMap<String, EventBusTransactionItem>(128);

    @Override
    public <T extends EventBus> T init() {
        if (!init.get()) {
            if (Objects.isNull(getEventStatusChangeManager())) {
                throw new NullPointerException("事件状态变更管理器为空");
            }
            if (Objects.isNull(getEventPersistenceManager())) {
                throw new NullPointerException("事件持久化管理器为空");
            }

            /* jvm 退出执行方法 */
            Runtime.getRuntime().addShutdownHook(new Thread(() -> DefaultEventBus.this.destroy()));
            getEventStatusChangeManager().init();
            getEventPersistenceManager().init();
            init.compareAndSet(Boolean.FALSE, Boolean.TRUE);
        } else {
            log.warn("不能重复执行初始化操作");
        }
        return (T) this;
    }

    @Override
    public <T extends EventBus, C extends EventBusConfig> T setConfig(C config) {
        this.config = config;
        return (T) this;
    }

    @Override
    public EventPersistenceManager getEventPersistenceManager() {
        if (this.eventPersistenceManager == null && this.config != null) {
            synchronized (this) {
                if (this.eventPersistenceManager == null && this.config != null) {
                    this.eventPersistenceManager = ExtensionLoader.getInstance(EventPersistenceManager.class)
                            .getExtensionOrDefault(config.getPersistenceManagerStrategyName());
                }
            }
        }
        return this.eventPersistenceManager;
    }

    @Override
    public <T extends EventBus> T setEventPersistenceManager(EventPersistenceManager eventPersistenceManager) {
        this.eventPersistenceManager = eventPersistenceManager;
        return (T) this;
    }

    @Override
    public EventChangeManager getEventStatusChangeManager() {
        if (this.eventStatusChangeManager == null && this.config != null) {
            synchronized (this) {
                if (this.eventStatusChangeManager == null && this.config != null) {
                    this.eventStatusChangeManager = ExtensionLoader.getInstance(EventChangeManager.class)
                            .getExtensionOrDefault(config.getEventChangeManagerStrategyName());
                }
            }
        }
        return this.eventStatusChangeManager;
    }

    @Override
    public <T extends EventBus> T setEventStatusChangeManager(EventChangeManager eventStatusChangeManager) {
        this.eventStatusChangeManager = eventStatusChangeManager;
        return (T) this;
    }

    @Override
    public <T extends EventBus> T registerEventHandler(@NonNull EventConsumerConfig eventConsumerConfig, @NonNull EventHandler eventHandler, EventRetryStrategy eventRetryStrategy) {

        EventConsumer eventConsumer = ExtensionLoader
                .getInstance(EventConsumer.class)
                .newExtension(eventConsumerConfig.getEventConsumerStrategyName(), eventConsumerConfig.getConstructorArgs())
                .setConfig(eventConsumerConfig)
                .setEventBus(this)
                .setEventHandler(eventHandler)
                .setEventRetryStrategy(eventRetryStrategy);

        Queue<EventConsumer> queue = getChannelProvider(eventConsumerConfig.getEventSourceName()).getConsumerQueue();
        queue.offer(eventConsumer);
        eventConsumer.start();
        return (T) this;
    }

    @Override
    public <T extends EventBus> T publishEventProducer(@NonNull EventProducerConfig eventProducerConfig) {
        ChannelProvider channelProvider = getChannelProvider(eventProducerConfig.getEventSourceName());

        if (!channelProvider.getEventProducerReference().isMarked()) {

            EventProducer eventProducer = ExtensionLoader
                    .getInstance(EventProducer.class)
                    .newExtension(eventProducerConfig.getEventProducerStrategyName(), eventProducerConfig.getConstructorArgs())
                    .setConfig(eventProducerConfig)
                    .init();

            channelProvider.getEventProducerReference().compareAndSet(null, eventProducer, false, true);
        }
        return (T) this;
    }

    @Override
    public <V> EventResult<V> sendEvent(@NonNull Event event) {
        checkup();
        try {
            EventResult result = getChannelProvider(event.getEventSourceName())
                    .getEventProducerReference()
                    .getReference()
                    .sendEvent(event);
            event.setStatus(Event.Status.RESULTED);
            return result;
        } catch (Exception e) {
            event.setStatus(Event.Status.RESULT_FAILURED);
            throw new BusinessException("事件处理异常", e);
        }
    }

    @Override
    public <V> void asyncSendEvent(@NonNull Event event, Callback<EventResult<V>, Void> callback) {
        checkup();
        if (callback == null) {
            sendAsyncEvent(event);
        } else {
            try {
                getChannelProvider(event.getEventSourceName())
                        .getEventProducerReference()
                        .getReference()
                        .asyncSendEvent(new CallbackEvent<>(event, callback));
            } catch (Exception e) {
                event.setStatus(Event.Status.RESULT_FAILURED);
                throw new BusinessException("事件处理异常", e);
            }
        }
    }

    @Override
    public void sendAsyncEvent(@NonNull Event event) {
        checkup();
        try {
            getChannelProvider(event.getEventSourceName())
                    .getEventProducerReference()
                    .getReference()
                    .sendAsyncEvent(event);
        } catch (Exception e) {
            event.setStatus(Event.Status.SEND_FAILURED);
            throw new EventSendException("事件发送异常", e);
        }
    }

    @Override
    public void batchSendAsyncEvent(@NonNull List<Event> eventList) {
        checkup();
        eventList.stream()
                .collect(Collectors.groupingBy(event -> event.getEventSourceName(), HashMap::new, Collectors.toList()))
                .entrySet()
                .parallelStream()
                .forEach((entry) -> {
                    List<Event> channelEventList = entry.getValue();
                    if (!(channelEventList == null || channelEventList.isEmpty())) {
                        try {
                            /* 批量发送 */
                            getChannelProvider(entry.getKey())
                                    .getEventProducerReference()
                                    .getReference()
                                    .batchSendAsyncEvent(channelEventList);
                        } catch (Exception e) {
                            channelEventList.forEach(event -> event.setStatus(Event.Status.SEND_FAILURED));
                            throw new EventSendException("事件发送异常", e);
                        }
                    }
                });
    }

    @Override
    public String newTransaction() {
        checkup();
        ExtensionLoader<IdGeneratorStrategy> idGeneratorStrategyExtensionLoader = ExtensionLoader.getInstance(IdGeneratorStrategy.class);
        // 获取配置的Id生成策略
        String transactionId = String.valueOf(idGeneratorStrategyExtensionLoader.getExtensionOrDefault(getConfig().getTransactionIdStrategyName()).generate());
        if (getEventTransactionMap().containsKey(transactionId)) {
            return newTransaction();
        } else {
            getEventTransactionMap().putIfAbsent(transactionId, new DefaultEventBusTransactionItemImpl(transactionId));
        }
        return transactionId;
    }

    @Override
    public void addEvent(@NonNull String transactionId, @NonNull Event... events) {
        checkup();
        getEventBusTransactionItem(transactionId).addEvent(events);
    }

    @Override
    public boolean commit(@NonNull String transactionId) {
        checkup();
        try {
            return getEventBusTransactionItem(transactionId).commit();
        } finally {
            cleanEventBusTransactionItem(transactionId);
        }
    }

    @Override
    public void rollback(@NonNull String transactionId) {
        checkup();
        try {
            getEventBusTransactionItem(transactionId).rollback();
        } finally {
            cleanEventBusTransactionItem(transactionId);
        }
    }

    @Override
    public void destroy() {
        if (this.init.get()) {
            this.init.compareAndSet(Boolean.TRUE, Boolean.FALSE);

            getEventTransactionMap()
                    .entrySet()
                    .parallelStream()
                    .forEach(entry -> entry.getValue().rollback());

            getEventTransactionMap().clear();
            if (Objects.nonNull(getEventStatusChangeManager())) {
                getEventStatusChangeManager().destroy();
            }
            if (Objects.nonNull(getEventPersistenceManager())) {
                getEventPersistenceManager().destroy();
            }
        } else {
            log.warn("不能重复执行销毁操作");
        }
    }

    /**
     * 验证
     */
    private void checkup() {
        if (Objects.isNull(getConfig())) {
            throw new NullPointerException("事件总线相关配置信息不能为空");
        }
        if (!this.init.get()) {
            throw new RuntimeException("请先执行初始化相关操作");
        }
    }

    /**
     * 获取事件执行通道信息
     *
     * @param eventSourceName
     */
    @NonNull
    private ChannelProvider getChannelProvider(@NonNull String eventSourceName) {
        ChannelProvider channelProvider = getChannelProviderMap().get(eventSourceName);
        if (channelProvider == null) {
            getChannelProviderMap().putIfAbsent(eventSourceName, new ChannelProvider(eventSourceName));
            channelProvider = getChannelProviderMap().get(eventSourceName);
        }
        return channelProvider;
    }

    /**
     * 获取事物项
     *
     * @param transactionId
     * @return
     */
    private EventBusTransactionItem getEventBusTransactionItem(@NonNull String transactionId) {
        EventBusTransactionItem transactionItem = getEventTransactionMap().get(transactionId);
        if (transactionItem == null) {
            throw new IllegalStateException("the specified transaction is not turned on or has been executed. please open a new transaction first . ");
        }
        return transactionItem;
    }

    private void cleanEventBusTransactionItem(@NonNull String transactionId) {
        getEventTransactionMap().remove(transactionId);
    }


    /**
     * 事件执行通道信息
     */
    @Getter
    private class ChannelProvider {
        /**
         * 执行通道标识
         */
        private final String eventSourceName;
        /**
         * 目前支持一个事件执行通道多个消费者
         */
        private final Queue<EventConsumer> consumerQueue = new ConcurrentLinkedQueue<>();
        /**
         * 目前仅支持一个事件执行通道一个生产者
         */
        private final AtomicMarkableReference<EventProducer> eventProducerReference = new AtomicMarkableReference(null, Boolean.FALSE);

        public ChannelProvider(final String eventSourceName) {
            this.eventSourceName = eventSourceName;
        }
    }

    /**
     * 默认事物管理器
     */
    private class DefaultEventBusTransactionItemImpl implements EventBusTransactionItem {

        /**
         * 事物项标识
         */
        @Getter
        private final String transactionId;
        /**
         * 是否已提交成功
         */
        private final AtomicBoolean isCommited;
        /**
         * 是否已执行回滚操作
         */
        private final AtomicBoolean isRollback;
        /**
         * 子事物管理器信息
         */
        @Getter
        private final ConcurrentMap<String, EventProducerTransactionItem> eventProducerTransactionItemMap;
        /**
         * 事物事件队列
         */
        @Getter
        private final AggregationEvent aggregationEvent = new AggregationEvent.EventBuilder().build();

        public DefaultEventBusTransactionItemImpl(@NonNull final String transactionId) {
            this.transactionId = transactionId;
            this.isCommited = new AtomicBoolean();
            this.isRollback = new AtomicBoolean();
            this.eventProducerTransactionItemMap = new ConcurrentHashMap<>(16);
        }

        @Override
        public void addEvent(Event... events) {
            if (events == null || events.length == 0) {
                return;
            }
            for (Event event : events) {
                if (event != null) {
                    final String eventSourceName = event.getEventSourceName();
                    EventProducerTransactionItem channelTransactionItem = getEventProducerTransactionItemMap().get(eventSourceName);
                    if (channelTransactionItem == null) {
                        EventProducerTransactionItem newChannelTransactionItem = Optional.ofNullable(getChannelProviderMap().get(eventSourceName))
                                .orElseThrow(() -> new NullPointerException("事件源" + eventSourceName + "不存在"))
                                .getEventProducerReference()
                                .getReference()
                                .newTransaction();

                        getEventProducerTransactionItemMap().putIfAbsent(eventSourceName, newChannelTransactionItem);
                        channelTransactionItem = getEventProducerTransactionItemMap().get(eventSourceName);
                    }
                    getAggregationEvent().addEvent(event);
                    channelTransactionItem.addEvent(event);
                }
            }
        }

        @Override
        public void rollback() {
            if (this.isCommited.get()) {
                throw new IllegalStateException("事务管理器已执行提交操作，不能再执行回滚操作了。");
            }
            if (!this.isRollback.get()) {
                this.isRollback.compareAndSet(Boolean.FALSE, Boolean.TRUE);

                getAggregationEvent().rollback();
                getEventProducerTransactionItemMap().entrySet()
                        .stream()
                        .forEach(entry -> entry.getValue().rollback());
            }
        }

        @Override
        public boolean commit() {
            if (this.isRollback.get()) {
                throw new IllegalStateException("事务管理器已执行回滚操作，不能再执行提交操作了。");
            }
            if (!this.isCommited.get()) {
                this.isCommited.compareAndSet(Boolean.FALSE, Boolean.TRUE);

                try {
                    getAggregationEvent().setSendType(Event.SendType.ASYNC_TRANSACTION);
                    getEventProducerTransactionItemMap().entrySet()
                            .parallelStream()
                            .forEach(entry -> entry.getValue().commit());
                    getAggregationEvent().commit();
                    return Boolean.TRUE;
                } catch (Exception e) {
                    log.error("事务提交异常", e);
                    getAggregationEvent().rollback();
                    return Boolean.FALSE;
                }
            }
            return Boolean.TRUE;
        }
    }
}
